package com.mercury.kafka.interceptor;

import com.mercury.kafka.domain.KafKaMqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * kafka拦截器
 * 生产者拦截器
 *
 * @author mercury
 */

@Slf4j
public class KafkaProducerInterceptor implements ProducerInterceptor<String, KafKaMqMessage> {

    /**
     * 发送成功计数
     */
    private volatile Integer success = 0;

    /**
     * 发送失败计数
     */
    private volatile Integer failure = 0;

    /**
     *
     */
    @Override
    public void configure(Map<String, ?> configs) {

    }

    /**
     * 发送消息已经操作消息的方法
     */
    @Override
    public ProducerRecord<String, KafKaMqMessage> onSend(ProducerRecord<String, KafKaMqMessage> record) {
        return new ProducerRecord<>(
                // 主题
                record.topic(),
                // 分区
                record.partition(),
                // 时间戳
                record.timestamp(),
                // key值
                record.key(),
                record.value(),
                record.headers());
    }

    /**
     * ack确认的方法
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        synchronized (this) {
            if (exception == null) {
                success++;
            } else {
                failure++;
            }
        }
    }


    /**
     * 关闭的方法，发送成功之后会将拦截器关闭，调用此方法
     */
    @Override
    public void close() {
        double successRation = (double) success / (success + failure);
        log.info("发送成功率： " + String.format("%f", successRation * 100) + "%");
    }


}